/*
 * This file is part of b.AMP.
 *
 * b.AMP is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * b.AMP is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with b.AMP.  If not, see <http://www.gnu.org/licenses/>.
 */


#include "async_loop.hpp"
#include "message_ack.hpp"
#include "message_source.hpp"
#include "message_target.hpp"
#include "process.hpp"

#include <bamp/message_ptr.hpp>

#include <boost/filesystem.hpp>
#include <boost/functional.hpp>
#include <boost/iostreams/stream.hpp>
#include <boost/iostreams/stream_buffer.hpp>
#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/system/error_code.hpp>

#include <iostream>
#include <memory>
#include <vector>

#include <sys/types.h>
#include <sys/wait.h>
#include <signal.h>
#include <unistd.h>


namespace fs  = boost::filesystem;
namespace io  = boost::iostreams;
namespace sys = boost::system;


/*
 * Implementation.
 */
struct bamp::Process::Implementation
{
    Implementation(Descriptor desc, Serializer ser, Deserializer des)
        : m_ser(ser),
          m_des(des),
          m_target(NULL),
          m_pid(-1)
    {
        // Copy the executable name to a C string.
        {
            const char*  exec = desc.exec.c_str();
            const size_t size = desc.exec.string().size() + 1;
            m_exec = (char*)malloc(size * sizeof(char));
            strcpy(m_exec, exec);
        }

        // Copy the argument list into a null-terminated array of C
        // strings.
        {
            const size_t argc = desc.args.size() + 1;
            m_argv = (char**)malloc(argc * sizeof(char*));
            for (size_t i = 0; i < argc-1; ++i)
            {
                const char*  arg  = desc.args[i].c_str();
                const size_t size = desc.args[i].size() + 1;
                m_argv[i] = (char*)malloc(size * sizeof(char*));
                strcpy(m_argv[i], arg);
            }
            m_argv[argc-1] = NULL;
        }
    }

    ~Implementation()
    {
        // Free the resources used by the internal C strings.
        for (size_t i = 0; m_argv[i] != NULL; ++i)
        {
            free(m_argv[i]);
        }
        free(m_argv);
        free(m_exec);
    }

    Serializer     m_ser;
    Deserializer   m_des;
    MessageTarget* m_target;
    char*          m_exec;   // Executable filename
    char**         m_argv;   // Arguments used to invoke the program

    // Child process
    pid_t m_pid;
    int   m_stdin;
    int   m_stdout;
};


/*
 * Public interface.
 */
 
bamp::Process::Process(const Descriptor desc, Serializer ser, Deserializer des)
    : pimpl(new Implementation(desc, ser, des))
{
    BAMP_ASSERT_THROW(fs::exists(pimpl->m_exec), NOT_FOUND)
}


bamp::Process::~Process()
{
    // Nothing here
}


void bamp::Process::setTarget(MessageTarget& target)
{
    pimpl->m_target = &target;
}


void bamp::Process::clearTarget()
{
    pimpl->m_target = NULL;
}


void bamp::Process::send(const Message& message, MessageAck ack)
{
    // Decrement the message's reference count
    ack.ack();

    // Only proceed if there is a process to write to
    if (pimpl->m_pid == -1) return;

    // Serialize the message to the process' stdin.
    {
        io::file_descriptor_sink sink(pimpl->m_stdin, io::never_close_handle);
        io::stream_buffer<io::file_descriptor_sink> buf(sink);
        std::ostream stream(&buf);
        pimpl->m_ser(stream, message);
    }
}


bamp::MessageAck bamp::Process::send(const Message& message)
{
    MessageAck ack(1);
    send(message, ack);
    return ack;
}


bool bamp::Process::preRun()
{
    // Create pipes to communicate with the new child process.
    int in[2];
    int out[2];

    if (pipe(in) < 0)
    {
        throw sys::system_error(errno, sys::system_category());
    }
    if (pipe(out) < 0)
    {
        close(in[0]);
        close(in[1]);
        throw sys::system_error(errno, sys::system_category());
    }

    // Fork and execute the program in the new child process.
    pid_t pid;
    switch(pid = fork())
    {
    case -1:
        close(out[0]);
        close(out[1]);
        close(in[0]);
        close(in[1]);
        throw sys::system_error(errno, sys::system_category());

    case 0:
        // New child process here. Redirect stdin and stdout to
        // internal pipes and run the program.
        close(out[0]);
        dup2(out[1], STDOUT_FILENO);
        close(out[1]);

        close(in[1]);
        dup2(in[0],  STDIN_FILENO);
        close(in[0]);

        if (execvp(pimpl->m_exec, pimpl->m_argv) < 0)
        {
            exit(1);
        }
    }

    // Parent process here. Close stdout's write end and stdin's read
    // end. The two other ends of the pipe will be used to communicate
    // with the child process.
    close(out[1]);
    close(in[0]);

    pimpl->m_pid    = pid;
    pimpl->m_stdin  = in[1];
    pimpl->m_stdout = out[0];

    return true;
}


bool bamp::Process::postRun()
{
    // Nothing here
    return true;
}


bool bamp::Process::iteration()
{
    // Deserialize messages from the process' stdout
    io::file_descriptor_source src(pimpl->m_stdout, io::never_close_handle);
    io::stream_buffer<io::file_descriptor_source> buf(src);
    std::istream stream(&buf);
    pimpl->m_des(stream, *pimpl->m_target);

    return true;
}


bool bamp::Process::preStop()
{
    // Tell the process to stop by closing the pipes and sending
    // sigterm.
    close(pimpl->m_stdin);
    close(pimpl->m_stdout);
    if (kill(pimpl->m_pid, SIGTERM) == -1)
    {
        throw sys::system_error(errno, sys::system_category());
    }
    return true;
}


void bamp::Process::postStop()
{
    int status;
    if (waitpid(pimpl->m_pid, &status, 0) == -1)
    {
        throw sys::system_error(errno, sys::system_category());
    }
}
